package com.mq.mqserver.core;
/*
* 通过这个类，来实现消费消息的核心逻辑*/

import com.mq.common.Consumer;
import com.mq.common.ConsumerEnv;
import com.mq.common.exception.MqException;
import com.mq.mqserver.VirtualHost;

import java.util.concurrent.*;

public class ConsumerManager {
    //持有上层的 VirtualHost 对象的引用，用来操作数据
    private VirtualHost parent;
    //只定一个线程池，负责执行 具体的回调任务
    private ExecutorService workerPool = Executors.newFixedThreadPool(4);
    //存放令牌的队列
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();
    //扫描线程
    private Thread scannerThread = null;

    public ConsumerManager(VirtualHost p){
        parent = p;

        scannerThread = new Thread(()->{
            while(true){
                try{
                    //1.拿到令牌
                    String queueName = tokenQueue.take();
                    //2.根据令牌，找到队列
                    MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                    if(queue == null){
                        throw new MqException("[ConsumerManager] 取令牌之后发现，该队列名不存在 queueName :" + queueName);

                    }
                    //3.从这个队列中消费一个消息
                    synchronized (queue){
                        consumeMessage(queue);
                    }
                } catch (MqException | InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        //把线程设为后台线程
        //后台线程：不会影响前代线程，前台线程结束，后台线程自然而然会自己结束
        scannerThread.setDaemon(true);
        scannerThread.start();
    }
    //这个方法调用的时机就是发送消息的时候
    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    public void addConsumer(String consumerTag, String queueName, boolean autoAk, Consumer consumer) throws MqException {
        //找到对应的队列
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if(queue == null){
            throw new MqException("[ConsumerManager] 队列不存在！ queueName = " + queueName);
        }
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag,queueName,autoAk,consumer);
        synchronized (queue){
            queue.addConsunerEvn(consumerEnv);
            //如果当前队列中已经有了一些消息，需要立即消费掉
            int n = parent.getMemoryDataCenter().getMessageCount(queueName);
            for(int i = 0; i < n; i++){
                //这个方法调用一次就消费一条消息
                consumeMessage(queue);
            }
        }
    }

    private void consumeMessage(MSGQueue queue) {
        //1.按照轮询的方式，找个消费者出来
        ConsumerEnv luyDog =  queue.chooseConsumer();
        if(luyDog == null){
            //没有消费者。暂时不消费，等有消费者出现再说
            return;
        }
        //2.从队列中取出来一个消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if(message == null){
            //当前队列中没有消息，无需消费
            return;
        }
        CountDownLatch latch = new CountDownLatch(1); // 添加同步点
        System.out.println("[ConsumerManager] 消息被成功消费! queueName=" +
                queue.getName() + ", messageId=" + message.getMessageId());
        //3.把消息带入到消费者的回调方法中，丢给线程池执行
        workerPool.submit(()->{
            try {
                //1.把消息放到待确认的集合中
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);
                //2.执行回调
                luyDog.getConsumer().handleDelivery(luyDog.getConumerTag(), message.getBasicProperties(),
                        message.getBody());
                //3.如果当前是”自动应答“，就可以直接删除消息
                // 如果当前是”手动应答“。则不处理，交给消费者调用 basicAck 处理
                if (luyDog.isAutoAck()) {
                    //1)删除硬盘上的消息
                    if (message.getDeliverMode() == 2) {
                        parent.getDiskDataCenter().deleteMessage(queue, message);
                    }
                    //2) 删除待确认集合的消息
                    parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(),message.getMessageId());
                    //3)删除内存中消息中心的消息
                    parent.getMemoryDataCenter().deleteMessage(message.getMessageId());
                    System.out.println("[ConsumerManager] 消息被成功消费！queueName = "+ queue.getName());

                }
            }catch (Exception e){
                e.printStackTrace();
            }finally {
                latch.countDown(); // 任务完成时计数减一
            }
            System.out.println("[ConsumerManager] queeuName=" + queue.getName());
        });
        try {
            latch.await(); // 等待任务完成
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

}
